package com.bckj.fastboot.iot.datcol.mqtt.service.impl;

import com.bckj.fastboot.iot.datcol.mqtt.config.MqttProperties;
import com.bckj.fastboot.iot.datcol.mqtt.service.IotMqttService;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Service;

/**
 * IotMqttServiceImpl
 *
 * @author wfc
 **/
@Slf4j
@Service
public class IotMqttServiceImpl implements IotMqttService {

    private final MqttProperties mqttProperties;
    private final MqttCallback mqttCallback;
    private MqttAsyncClient mqttClient;

    public IotMqttServiceImpl(MqttProperties mqttProperties, MqttCallback mqttCallback) {
        this.mqttProperties = mqttProperties;
        this.mqttCallback = mqttCallback;
    }

    @PostConstruct
    public void init() {
        try {
            // MemoryPersistence 设置为内存存储，也可以使用 FilePersistence 持久化
            mqttClient = new MqttAsyncClient(
                    mqttProperties.getBrokerUrl(),
                    mqttProperties.getClientId(),
                    new MemoryPersistence()
            );

            // 设置回调
            mqttClient.setCallback(mqttCallback);

            // MQTT 连接选项
            MqttConnectOptions connOpts = getMqttConnectOptions();

            log.info("Connecting to MQTT broker: {}", mqttProperties.getBrokerUrl());
            mqttClient.connect(connOpts, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("MQTT Connected successfully!");
                    // 连接成功后订阅主题
                    subscribeToTopic(mqttProperties.getTopic());
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("MQTT Connection failed!", exception);
                }
            });

        } catch (MqttException e) {
            log.error("MQTT Client init error", e);
        }
    }

    private MqttConnectOptions getMqttConnectOptions() {
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true);
        if (mqttProperties.getUsername() != null && !mqttProperties.getUsername().isEmpty()) {
            connOpts.setUserName(mqttProperties.getUsername());
            connOpts.setPassword(mqttProperties.getPassword().toCharArray());
        }
        connOpts.setAutomaticReconnect(true); // 自动重连
        connOpts.setConnectionTimeout(10);
        connOpts.setKeepAliveInterval(60);
        return connOpts;
    }

    /**
     * 订阅主题
     */
    private void subscribeToTopic(String topic) {
        // 边缘网关
        try {
            mqttClient.subscribe("status/handle/#", 1);
        } catch (MqttException e) {
            log.error("Subscribe error", e);
        }
        // topics
        String[] ts = topic.split(",");
        try {
            int[] qos = new int[ts.length];
            mqttClient.subscribe(ts, qos, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("Subscribed to topics");
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("Subscription to topics failed", exception);
                }
            });
        } catch (MqttException e) {
            log.error("Subscribe error", e);
        }
    }

    /**
     * 发布消息到某个主题
     */
    @Override
    public void publish(String topic, String payload) {
        try {
            MqttMessage message = new MqttMessage(payload.getBytes());
            message.setQos(1);
            message.setRetained(false);
            mqttClient.publish(topic, message, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("Message published to topic {}: {}", topic, payload);
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("Failed to publish message to topic {}", topic, exception);
                }
            });
        } catch (MqttException e) {
            log.error("Publish error", e);
        }
    }

    @PreDestroy
    public void disconnect() {
        try {
            if (mqttClient != null && mqttClient.isConnected()) {
                mqttClient.disconnect();
                log.info("MQTT Disconnected gracefully.");
            }
        } catch (MqttException e) {
            log.error("Error while disconnecting MQTT", e);
        }
    }
}
